#pragma once
#include "requestor.hpp"
#include <unordered_set>
// 1.实现功能 : 创建, 删除, 订阅, 取消订阅, 发布消息
// 2.回调处理 : 给dispatcher注册消息处理的回调
// 3.关系维护 : (主题名称, 消息处理回调)
namespace jsonrpc
{
    namespace client
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            using SubCallback = std::function<void(const std::string &key, const std::string &msg)>;
            TopicManager(const Requestor::ptr &requestor):_requestor(requestor) {}
        public:
            // 1.功能的实现
            bool create(const BaseConnection::ptr &conn, std::string key) 
            {
                return commonRequest(conn, key, TopicOptype::TOPIC_CREATE);
            }
            bool remove(const BaseConnection::ptr &conn, const std::string &key) 
            {
                return commonRequest(conn, key, TopicOptype::TOPIC_REMOVE);
            }
            bool subscribe(const BaseConnection::ptr &conn, const std::string &key, const SubCallback &cb) 
            {
                addSubscribe(key, cb);
                bool ret = commonRequest(conn, key, TopicOptype::TOPIC_SUBSCRIBE);
                if(ret == false)
                {
                    delSubscribe(key);
                    return false;
                }
                return true;
            }
            bool cancel(const BaseConnection::ptr &conn, const std::string &key) 
            {
                delSubscribe(key);
                return commonRequest(conn, key, TopicOptype::TOPIC_CANCEL);
            }
            bool publish(const BaseConnection::ptr &conn, const std::string &key, const std::string &msg)
            {
                return commonRequest(conn, key, TopicOptype::TOPIC_PUBLISH, msg);
            }

            // 2.回调的处理 -- 注册给dispatcher
            void onPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 对消息类型做检查
                auto optype = msg->optype();
                if(optype != TopicOptype::TOPIC_PUBLISH)
                {
                    ELOG("收到错误的主题操作类型");
                    return;
                }
                // 拿出key + msg --> 找到对应callback --> 调用
                std::string topic_key = msg->topicKey();
                std::string topic_msg = msg->topicMsg();
                auto topic_cb = getSubscribe(topic_key);
                if(topic_cb == nullptr)
                {
                    ELOG("该主题类型没有对应的回调函数:%s", topic_key.c_str());
                    return;
                }
                return topic_cb(topic_key, topic_msg);
            }
        private:
            // 构建请求
            bool commonRequest(const BaseConnection::ptr& conn, const std::string& key, TopicOptype optype, const std::string &message = " ")
            {
                // 构建请求对象 + 填充数据
                auto msg_req = MessageFactory::create<TopicRequest>();
                msg_req->setId(UUID::uuid());
                msg_req->setMType(MType::REQ_TOPIC);
                msg_req->setOptype(optype);
                msg_req->setTopicKey(key);
                if(optype == TopicOptype::TOPIC_PUBLISH){
                    msg_req->setTopicMsg(message);
                }
                // 向服务端发送请求
                BaseMessage::ptr msg_rsp;
                bool ret = _requestor->send(conn, msg_req, msg_rsp);
                if(ret == false)
                {
                    ELOG("主题操作请求失败");
                    return false;
                }
                // 判断请求处理是否成功
                auto topic_msg_rsp = std::dynamic_pointer_cast<TopicResponse>(msg_rsp);
                if(topic_msg_rsp == nullptr)
                {
                    ELOG("类型转换失败");
                    return false;
                }
                if(topic_msg_rsp->rcode() != RCode::RCODE_OK)
                {
                    ELOG("主题操作请求出错");
                    return false;
                }
                return true;
            }

            // 3.关系的维护 -- 主题订阅的增删查
            void addSubscribe(const std::string& key, const SubCallback &cb)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.insert(std::make_pair(key, cb));
            }
            void delSubscribe(const std::string& key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.erase(key);
            }
            const SubCallback getSubscribe(const std::string& key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto cb_it = _topic_callbacks.find(key);
                if(cb_it == _topic_callbacks.end())
                {
                    ELOG("没有该主题所对应的回调");
                    return SubCallback();
                }
                return cb_it->second;
            }
        private:
            std::mutex _mutex;
            std::unordered_map<std::string, SubCallback> _topic_callbacks; 
            Requestor::ptr _requestor;
        };
    }
}